package com.galudisu

import akka.actor.SupervisorStrategy._
import akka.actor.{ActorSystem, OneForOneStrategy, Props}
import akka.pattern.ask
import akka.routing.{DefaultResizer, Resizer, RoundRobinPool}
import akka.util.Timeout
import com.galudisu.jooq.PoolableFactory

import scala.concurrent.Future
import scala.concurrent.duration.Duration
import scala.util.{Failure, Success}

class WorkerPool[T](
                     val name: String,
                     val size: Int,
                     val defaultTimeout: Timeout,
                     val maxNrOfRetries: Int,
                     val retryRange: Duration,
                     val objectFactory: PoolableFactory[T])(implicit actorSystem: ActorSystem) {

  private val supervisor =
    OneForOneStrategy(
      maxNrOfRetries = maxNrOfRetries,
      withinTimeRange = retryRange) {

      case _: Throwable => Restart
    }

  private val router =
    actorSystem.actorOf(
      props =
        Props(classOf[Worker[T]], objectFactory)
        .withRouter(RoundRobinPool(
          nrOfInstances = size,
          supervisorStrategy = supervisor)),
      name = name)

  import actorSystem.dispatcher

  def execute[U](fn: T => U)(implicit timeout: Timeout = defaultTimeout): Future[U] =
    ask(router, Job[T, U](fn)).map {
      case Success(res: Any) => res.asInstanceOf[U]
      case Failure(t) => throw new AsyncPoolsException("AsyncPools execution error", t)
    }

}